//
// Created by xbm on 23-3-10.
//

#include "handler.h"
#include <thread>
#include "fmt/format.h"
#include "log.h"
#include "../timer/timer.h"
#include "../socket/epoll.h"
#include "../socket/socketServer.h"
#include "../socket/socketUtil.h"
#include "../dateBase/dateBase.h"
#include "dispatcher.h"
#include "eventLoop.h"

using namespace std;
using namespace socket_epoll;
using namespace timer;
using namespace dateBase;

namespace event {


void ResultSupervisorHandler::addResult(future<void> f) {
    results_.push(std::move(f));
}

ResultSupervisorHandler::ResultSupervisorHandler() {
    log("resultSupervisorHandler start running!");

}

ResultSupervisorHandler::~ResultSupervisorHandler() {
    log("resultSupervisorHandler end running!");
    close();
}

void ResultSupervisorHandler::working() {
    while (running_) {
        constexpr auto SLEEP_TIME = 1s;
        this_thread::sleep_for(SLEEP_TIME);

        int sz = results_.size();
        log(fmt::format("handling results, current have {} tasks", sz));
        int cur = 0;
        int err = 0;
        while (auto f = results_.pop()) {
            try {
                if (f->wait_for(0s) != std::future_status::ready) {
                    ++suspends_;
                    results_.push(std::move(*f));
                    continue;
                }
                f->get();
                ++finished_;
                ++cur;
            }
            catch (const future_error &e) {
                ++errors_;
                log(fmt::format("caught a future_error with code {}\nMessage: {}", e.code().value(), e.what()));
            }
            catch (const runtime_error &e) {
                ++errors_;
                log(fmt::format("caught a future_error with code {}\n", e.what()));
            }
        }
        log(fmt::format("handle results finished! {}/{} handled, {} errors occurred", cur, sz, err));
        log(fmt::format("all state: finished {}, errored {}, suspend {}", finished_, errors_, suspends_));
    }
}

void ResultSupervisorHandler::close() {
    running_ = false;
}

void ResultSupervisorHandler::run() {
    thread t(&ResultSupervisorHandler::working, this);
    t.detach();
}


void runSelfEventHandler(shared_ptr<Event> e) {
    static_pointer_cast<RunSelfEvent>(e)->runner();
}

void registerHandles() {
    auto getNum = [](int16_t major, int16_t minor) { return Event_Type{major, minor}.num; };
    Dispatcher &dispatcher = Dispatcher::getInstance();
    dispatcher.registerHandler(getNum(RunSelfEvent::MAJOR_TYPE, 0), runSelfEventHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::ACCEPT), socketAcceptHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::READ), socketReadHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::WRITE), socketWriteHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::ERROR), socketErrorHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::CLOSE), socketCloseHandler);
    dispatcher.registerHandler(getNum(SocketEvent::MAJOR_TYPE, SocketEvent::EXPIRE), socketExpireHandler);
}

auto getTimestamp() {
    return chrono::high_resolution_clock::now();
}

constexpr int DEFAULT_STATE = static_cast<int>(Epoll::EVENT_TYPE::HUP) | static_cast<int>(Epoll::EVENT_TYPE::ERR) |
                              static_cast<int>(Epoll::EVENT_TYPE::ET) | static_cast<int>(Epoll::EVENT_TYPE::ONE_SHOT);

void setExpireTimer(int fd) {
    timer::Timer::getInstance().addTimer_for(SocketEvent::EXPIRE_TIME, [fd](const timer::Time &) {
        EventLoop::getInstance().addEvent(SocketEvent::make_event(fd, SocketEvent::EXPIRE));
    });
}

void socketAcceptHandler(std::shared_ptr<Event> e) {
    static Epoll &epoll = Epoll::getInstance();
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    while (auto &&client = SocketServer::getInstance().accept()) {
        client->setNonBlocking();
        setExpireTimer(client->getFD());
        SocketEvent::fd2status[client->getFD()] = {getTimestamp(), *client};
        Epoll::getInstance().add(client->getFD(), static_cast<int>(Epoll::EVENT_TYPE::IN) | DEFAULT_STATE);
    }

}

void socketReadHandler(std::shared_ptr<Event> e) {
    static Epoll &epoll = Epoll::getInstance();
    static DateBase &dateBase = DateBase::getInstance();
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    setExpireTimer(event->fd);
    SocketEvent::Status &status = SocketEvent::fd2status[event->fd];
    status.timestamp = getTimestamp();
    status.socketClient.init(event->fd);
    string ret = status.socketClient.read();
    if (ret.empty()) {
        EventLoop &eventLoop = EventLoop::getInstance();
        eventLoop.addEvent(SocketEvent::make_event(event->fd, SocketEvent::CLOSE));
        return;
    }
    status.buffer += ret;

    if (status.buffer.ends_with("\r\n")) {
        stringstream ss(status.buffer);
        string cmd, k, v;
        ss >> cmd >> k >> v;
        if (cmd == "set")status.buffer = dateBase.set(k, v);
        else if (cmd == "get") status.buffer = dateBase.get(k);
        epoll.mod(event->fd, static_cast<int>(Epoll::EVENT_TYPE::OUT) | DEFAULT_STATE);
    }
    else {
        epoll.mod(event->fd, static_cast<int>(Epoll::EVENT_TYPE::IN) | DEFAULT_STATE);

    }
}

void socketWriteHandler(std::shared_ptr<Event> e) {
    static Epoll &epoll = Epoll::getInstance();
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    SocketEvent::Status &status = SocketEvent::fd2status[event->fd];
    status.socketClient.init(event->fd);
    status.socketClient.write(status.buffer + "\r\n");
    status.buffer.clear();
    epoll.mod(event->fd, static_cast<int>(Epoll::EVENT_TYPE::IN) | DEFAULT_STATE);
}

void socketErrorHandler(std::shared_ptr<Event> e) {
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    SocketEvent::Status &status = SocketEvent::fd2status[event->fd];
    EventLoop &eventLoop = EventLoop::getInstance();
    eventLoop.addEvent(SocketEvent::make_event(event->fd, SocketEvent::CLOSE));
}

void socketCloseHandler(std::shared_ptr<Event> e) {
    static Epoll &epoll = Epoll::getInstance();
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    SocketEvent::Status &status = SocketEvent::fd2status[event->fd];
    epoll.del(event->fd);
    status.socketClient.close();
    //SocketEvent::fd2status.erase(status.socketClient.getFD());

}

void socketExpireHandler(std::shared_ptr<Event> e) {
    shared_ptr<SocketEvent> event = static_pointer_cast<SocketEvent>(e);
    SocketEvent::Status &status = SocketEvent::fd2status[event->fd];
    if (status.timestamp + SocketEvent::EXPIRE_TIME <= chrono::high_resolution_clock::now()) {
        EventLoop &eventLoop = EventLoop::getInstance();
        eventLoop.addEvent(SocketEvent::make_event(event->fd, SocketEvent::CLOSE));
    }
}


}
